﻿/// Copyright © 2013 Microsoft Corporation. 
/// This project is released under the Microsoft Public License (MS-PL)
///
namespace DataMorph.Generator
{
    using System;
    using System.IO;
    using System.Threading;

    using DataMorph.Common.Contracts;
    using DataMorph.Common.Storage;
    
    using Microsoft.WindowsAzure.Storage.Blob;
    using Microsoft.WindowsAzure.Storage.RetryPolicies;

    public class BlobFileWriter : ITagDataFileWriter
    {
        private readonly ICloudStorageAccount cloudStorageAccount;

        private readonly string dataSetName;

        private readonly int day;

        private readonly CloudBlobContainer container;

        public BlobFileWriter(ICloudStorageAccount cloudStorageAccount, string dataSetName, int day)
        {
            this.cloudStorageAccount = cloudStorageAccount;
            this.dataSetName = dataSetName.ToLower();
            this.day = day;

            var cloudBlobClient = this.cloudStorageAccount.CreateCloudBlobClient();

            int completionPortThreads;
            int workerThreads;
            ThreadPool.GetMinThreads(out workerThreads, out completionPortThreads);

            cloudBlobClient.ParallelOperationThreadCount = completionPortThreads;

            this.container = cloudBlobClient.GetContainerReference(this.dataSetName);
            this.container.CreateIfNotExists();
        }

        public string Write(TagDataFile file)
        {
            var blockBlob = this.container.GetBlockBlobReference(string.Format("{0}/{1}", this.day, file.Id));
            using (var memoryStream = new MemoryStream())
            using (var streamWriter = new StreamWriter(memoryStream))
            {
                foreach (var tagData in file.Lines)
                {
                    streamWriter.Write(tagData.TagId);
                    streamWriter.Write(',');
                    streamWriter.Write(tagData.Ticks);
                    streamWriter.Write(',');
                    streamWriter.WriteLine(tagData.Value);
                }

                streamWriter.Flush();
                memoryStream.Seek(0, SeekOrigin.Begin);

                blockBlob.UploadFromStream(
                    memoryStream,
                    null,
                    new BlobRequestOptions { RetryPolicy = new ExponentialRetry(TimeSpan.FromSeconds(30), 5) });
            }

            return blockBlob.Uri.ToString();
        }
    }
}